package cn.baiyang.rocketmq.spring.autoconfigure.ratelimit;

import java.time.Duration;
import java.util.HashSet;
import java.util.List;
import java.util.Set;

import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeansException;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Conditional;
import org.springframework.context.annotation.Configuration;
import org.springframework.util.StringUtils;

import io.github.bucket4j.Bucket;
import io.github.bucket4j.distributed.ExpirationAfterWriteStrategy;
import io.github.bucket4j.distributed.proxy.ProxyManager;
import io.github.bucket4j.distributed.serialization.Mapper;
import io.github.bucket4j.redis.jedis.cas.JedisBasedProxyManager;
import redis.clients.jedis.HostAndPort;
import redis.clients.jedis.JedisCluster;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisSentinelPool;

/**
 *
 * @author hongzhu
 * @version V1.0
 * @since 2024-03-14 10:00
 */
@Configuration
@Conditional(RateLimitCondition.class)
@EnableConfigurationProperties(Bucket4jProperties.class)
@ConditionalOnClass({ Bucket.class})
public class Bucket4jAutoConfiguration implements ApplicationContextAware {
    private static final Logger LOGGER = LoggerFactory.getLogger(Bucket4jAutoConfiguration.class);

    private ApplicationContext applicationContext;

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }

    @Bean("jedisProxyManager")
    @ConditionalOnClass(JedisPool.class)
    @ConditionalOnProperty(prefix = "bucket4j", value = {"jedis.password"})
    public ProxyManager<String> jedisProxyManager(Bucket4jProperties properties) {
        if (null != properties.getRedisson()) {
            throw new RuntimeException("Either Jedis or Redisson Proxy Manager, but can't both.");
        }

        Bucket4jProperties.Jedis propertiesJedis = properties.getJedis();
        String user = propertiesJedis.getUser();
        String password = propertiesJedis.getPassword();
        JedisBasedProxyManager.JedisBasedProxyManagerBuilder<byte[]> jedisBasedProxyManagerBuilder;

        if (null != propertiesJedis.getSentinelNodes() && !propertiesJedis.getSentinelNodes().isEmpty()
            && !StringUtils.isEmpty(propertiesJedis.getMasterName())) {
            JedisSentinelPool jedisSentinelPool =  new JedisSentinelPool(
                propertiesJedis.getMasterName(), propertiesJedis.getSentinelNodes(), new GenericObjectPoolConfig(), 2000,
                StringUtils.isEmpty(password) ? null : password, propertiesJedis.getDatabase());
            jedisBasedProxyManagerBuilder = JedisBasedProxyManager.builderFor(jedisSentinelPool);

        } else if (null != propertiesJedis.getClusterNodes() && !propertiesJedis.getClusterNodes().isEmpty()) {
            Set<HostAndPort> jedisClusterNodes = new HashSet<HostAndPort>();
            List<String> propertiesJedisClusterNodes = propertiesJedis.getClusterNodes();
            for (String address : propertiesJedisClusterNodes) {
                String[] str = address.split(":");
                if (str.length != 2) {
                    throw new RuntimeException("Redis Address is illegal. Correct format is 'Host:Port'");
                }
                jedisClusterNodes.add(new HostAndPort(str[0], Integer.parseInt(str[1])));
            }
            JedisCluster jedisCluster = StringUtils.isEmpty(password) ?
                new JedisCluster(jedisClusterNodes, 2000, 2000, 5, new GenericObjectPoolConfig())
                : new JedisCluster(jedisClusterNodes, 2000, 2000, 5, password, new GenericObjectPoolConfig());
            jedisBasedProxyManagerBuilder = JedisBasedProxyManager.builderFor(jedisCluster);

        } else if (!StringUtils.isEmpty(propertiesJedis.getAddress())) {
            String address = propertiesJedis.getAddress();
            String[] str = address.split(":");
            if (str.length != 2) {
                throw new RuntimeException("Redis Address is illegal. Correct format is 'Host:Port'");
            }
            JedisPool jedisPool = StringUtils.isEmpty(password) ?
                new JedisPool(new GenericObjectPoolConfig(), str[0], Integer.parseInt(str[1]))
                : new JedisPool(new GenericObjectPoolConfig(), str[0], Integer.parseInt(str[1]), 2000, password, propertiesJedis.getDatabase());
            jedisBasedProxyManagerBuilder = JedisBasedProxyManager.builderFor(jedisPool);

        } else {
            throw new RuntimeException("No Jedis Configuration! Either using single redis server or redis cluster or redis sentinel mode");
        }

        JedisBasedProxyManager<String> jedisBasedProxyManager = jedisBasedProxyManagerBuilder.withExpirationStrategy(
            ExpirationAfterWriteStrategy.basedOnTimeForRefillingBucketUpToMax(Duration.ofSeconds(10))
            )
            .withKeyMapper(Mapper.STRING)
            .build();
        LOGGER.info("Successful to Create Bucket4j Jedis ProxyManager");
        return jedisBasedProxyManager;
    }

}
